package com.shujia.stream

import com.shujia.util.FlinkTableEnv
import org.apache.flink.table.api.TableEnvironment

object CardLocationToEs {
  def main(args: Array[String]): Unit = {

    val table: TableEnvironment = FlinkTableEnv.getTableEnv(5000)

    /**
      *
      * 卡口过车数据
      */
    table.executeSql(
      """
        |CREATE TABLE cars_kafka (
        |  `car` STRING,
        |  `city_code` STRING,
        |  `county_code` STRING,
        |  `card` BIGINT,
        |  `camera_id` STRING,
        |  `orientation` STRING,
        |  `road_id` BIGINT,
        |  `time` BIGINT,
        |  `speed`  DOUBLE,
        |  `ts` AS PROCTIME()
        |) WITH (
        |  'connector' = 'kafka',
        |  'topic' = 'cars',
        |  'properties.bootstrap.servers' = 'master:9092',
        |  'properties.group.id' = 'JIChaControl',
        |  'scan.startup.mode' = 'earliest-offset',
        |  'format' = 'json',
        |  'json.ignore-parse-errors' = 'true'
        |)
        |
      """.stripMargin)

    /**
      * 卡口维度表
      *
      */

    table.executeSql(
      """
        |CREATE TABLE dim_card_mysql (
        |  card BIGINT,
        |  lon DOUBLE,
        |  lat DOUBLE,
        |  PRIMARY KEY (card) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/car',
        |   'table-name' = 'dim_card',
        |   'username' = 'root',
        |   'password' = '123456'
        |);
      """.stripMargin)

    /**
      * 创建es sink表
      *
      */

    table.executeSql(
      """
        |CREATE TABLE cars_es (
        |  `car` STRING,
        |  `city_code` STRING,
        |  `county_code` STRING,
        |  `card` BIGINT,
        |  `camera_id` STRING,
        |  `orientation` STRING,
        |  `road_id` BIGINT,
        |  `time` BIGINT,
        |  `speed`  DOUBLE,
        |  `time_str`  STRING,
        |  `location` ARRAY<DOUBLE>,
        |  PRIMARY KEY (car,`time`) NOT ENFORCED
        |) WITH (
        |  'connector' = 'elasticsearch-7',
        |  'hosts' = 'http://master:9200',
        |  'index' = 'cars'
        |);
      """.stripMargin)


    /**
      * 关联卡口过车数据和卡口维度表获取卡口经纬度
      *
      */

    table.executeSql(
      """
        |
        |insert into cars_es
        |select
        |a.car,
        |a.city_code,
        |a.county_code,
        |a.card,
        |a.camera_id,
        |a.orientation,
        |a.road_id,
        |a.`time`,
        |a.speed,
        |FROM_UNIXTIME(a.`time`) as time_str,
        |ARRAY[b.lon,b.lat] as location
        | from
        |cars_kafka as a
        |inner join
        |dim_card_mysql  FOR SYSTEM_TIME AS OF a.ts AS b
        |on a.card=b.card
        |
        |
      """.stripMargin)

  }

}
